package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}

object Code17AggregateByKey {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("AggregateByKey"))

    // // 需求：统计班级各年龄段人数
    //    sc
    //      .textFile("spark_code/data/students.txt")
    //      .map{
    //        case line => {
    //          val splitRes: Array[String] = line.split(",")
    //          ((splitRes(4),splitRes(2)),1)
    //        }
    //      }
    //      .reduceByKey(_+_)
    //      .reduceByKey(
    //        (x,y) => {
    //          x + y
    //        }
    //      )
    /*
    (zeroValue: U) 第一个参数列表  zeroValue：初始值 对于每个Key都添加了一个初始的Value值  为了防止当前Key只有一个
    (seqOp: (U, V) => U, combOp: (U, U) => U) 第二个参数列表
          seqOp:函数  (U, V) => U:函数类型
              seqOp:函数 代表预计算过程，表示在Map端分区内进行做逻辑计算
          combOp:函数 (U, U) => U:函数类型
              combOp:函数 表示在Reduce端分区内进行做逻辑计算



    def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,
        combOp: (U, U) => U): RDD[(K, U)] = self.withScope

     */
    //      .aggregateByKey(0)(
    //        (first_value,next_value) => {first_value+next_value},
    //        (first_value,next_value) => {first_value+next_value}
    //      )
    //      .foreach(println)

    List(1, 2, 3, 6, 7, 8)

    sc
      .parallelize(
        List(("k1", 1), ("k1", 2), ("k1", 3), ("k1", 6), ("k1", 7), ("k1", 8))
        // 设置分区数为 2
        , 2
      )
      .mapPartitions {
        case iterator => {
//          println(iterator.toList)
          iterator
        }
      }
      /*
        对于得到的两个分区：
        List((k1,1), (k1,2), (k1,3)) =>  List((k1,0),(k1,1), (k1,2), (k1,3)) => (k1,3)
        List((k1,6), (k1,7), (k1,8)) =>  List((k1,0),(k1,6), (k1,7), (k1,8)) => (k1,8)  => (k1,11)
        对分区内进行求最大值，分区之间求和
       */
      .aggregateByKey(0)(
        (firstValue, nextValue) => {
          math.max(firstValue, nextValue)
        },
        (firstValue, nextValue) => {
          firstValue + nextValue
        }
      )
      .foreach(println)





  }
}
